package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/18
 *
 *      批处理用的场景少，Flink的特色，是对数据流进行有状态的计算！
 *
 *
 */
public class Demo1_BatchDemo
{
    public static void main(String[] args) throws Exception {

        //获取编程环境
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataSource<String> dataSource = executionEnvironment.readTextFile("data/words.txt");

        /*
                转换
                    ①每一行按照分隔符去切割
                    ②切割后的每个单词，都记录 （单词,1）
                        输入:  a b c  输出: (a,1)
                                           (b,1)
                                           (c,1)
                                 用flatMap


                               (t1,t2...): flink提供了专属的结构 Tuple
                                            根据元素的数量，如果只有2个元素，就是 Tuple2
                                            如果有3个元素: Tuple3
                                            最多只能有2x个元素，如果再多，只能自定义Bean封装
                    ③按照单词分组，对同组的1进行累加
         */
        FlatMapOperator<String, Tuple2<String, Integer>> ds = dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>()
        {
            /*
                String value: 输入的一行
                Collector<Tuple2<String, Integer>> out： 输出结果的收集器，调用方法将结果输出
             */
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {

                String[] words = line.split(" ");

                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }

            }
        });

        //按照tuple2第一个位置的数据分组
        AggregateOperator<Tuple2<String, Integer>> ds1 = ds.groupBy(0)
                                                           .sum(1);

        //sink：输出算子，输出到控制台
        ds1.print();


    }
}
